package com.partsoft.dits.integrate;

import org.apache.commons.collections.CollectionUtils;


import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import javax.sql.DataSource;

import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.Assert;

import com.partsoft.dits.TableDataEdgeDescriptor;
import com.partsoft.dits.dsm.DataSourceFactory;
import com.partsoft.dits.log.Log;
import com.partsoft.dits.message.DataProperty;
import com.partsoft.dits.message.MessageProcessor;
import com.partsoft.dits.transform.TransformReactor;

public class ReceiveMessageContext {
	
	private Map<String, DataSource> dataSources = new LinkedHashMap<String, DataSource>();
	
	private Map<String, List<TableDataEdgeDescriptor>> messageCodeReceives = new LinkedHashMap<String, List<TableDataEdgeDescriptor>>();
	
	private Map<TableDataEdgeDescriptor, String> receiveBeanTransformScripts = new LinkedHashMap<TableDataEdgeDescriptor, String>();
	
	private Map<TableDataEdgeDescriptor, DataProperty[]> receiveBeanTargetColumns = new LinkedHashMap<TableDataEdgeDescriptor, DataProperty[]>(); 
	
	private Map<PlatformTransactionManager, TransactionStatus> transactionStatuses;
	
	private final MessageProcessor messageProcessor;
	
	private final DataSourceFactory dataSourceFactory;
	
	private final TransformReactor transformReactor;

	ReceiveMessageContext(MessageProcessor messageProcessor, DataSourceFactory dataSourceManager, TransformReactor transformReactor) {
		Assert.notNull(messageProcessor);
		Assert.notNull(dataSourceManager);
		Assert.notNull(transformReactor);
		this.messageProcessor = messageProcessor;
		this.dataSourceFactory = dataSourceManager;
		this.transformReactor = transformReactor;
	}
	
	DataSource addDataSource(String dsName, DataSource ds) {
		Assert.hasText(dsName);
		Assert.notNull(ds);
		return dataSources.put(dsName, ds);
	}
	
	void addReceiveBean(String messageCode, TableDataEdgeDescriptor rBean) {
		Assert.hasText(messageCode);
		Assert.notNull(rBean);
		List<TableDataEdgeDescriptor> rbeans = messageCodeReceives.containsKey(messageCode) ? messageCodeReceives.get(messageCode) : new LinkedList<TableDataEdgeDescriptor>();
		rbeans.add(rBean);
		messageCodeReceives.put(messageCode, rbeans);
	}
	
	void addTransformScript(TableDataEdgeDescriptor bean, String script) {
		Assert.notNull(bean);
		Assert.hasText(script);
		receiveBeanTransformScripts.put(bean, script);
	}
	
	void addTableColumns(TableDataEdgeDescriptor bean, DataProperty clns[]) {
		Assert.notNull(bean);
		Assert.notEmpty(clns);
		receiveBeanTargetColumns.put(bean, clns);
	}
	
	void addReceiveBeans(String messageCode, List<TableDataEdgeDescriptor> rBean) {
		Assert.hasText(messageCode);
		Assert.notEmpty(rBean);
		List<TableDataEdgeDescriptor> rbeans = messageCodeReceives.containsKey(messageCode) ? messageCodeReceives.get(messageCode) : new LinkedList<TableDataEdgeDescriptor>();
		rbeans.addAll(rBean);
		messageCodeReceives.put(messageCode, rbeans);
	}
	
	public DataSource getDataSource(String dsName) {
		return dataSources.get(dsName);
	}
	
	public String getTransformScript(TableDataEdgeDescriptor bean) {
		return receiveBeanTransformScripts.get(bean);
	}
	
	@SuppressWarnings("unchecked")
	public Collection<TableDataEdgeDescriptor> getReceiveBeans(String messageCode) {
		return messageCodeReceives.containsKey(messageCode) ? messageCodeReceives.get(messageCode) : Collections.EMPTY_LIST;
	}
	
	public DataProperty[] getTableColumns(TableDataEdgeDescriptor bean) {
		return receiveBeanTargetColumns.get(bean);
	}
	
	public boolean hasTableColumns(TableDataEdgeDescriptor bean) {
		return receiveBeanTargetColumns.containsKey(bean);
	}
	
	public boolean hasReceiveDefined(String messageCode) {
		return messageCodeReceives.containsKey(messageCode) ? (CollectionUtils.isEmpty(messageCodeReceives.get(messageCode)) == false) : false;
	}
	
	public boolean hasDataSource(String dsName) {
		return dataSources.containsKey(dsName);
	}
	
	public Map<String, DataSource> getDataSources() {
		return new HashMap<String, DataSource>(dataSources);
	}
	
	Map<String, DataSource> getRawDataSources() {
		return dataSources;
	}
	
	public void startContextTransaction() {
		if (transactionStatuses != null)
			throw new IllegalStateException("已开启事务，不能重新开启!");
		if (Log.isDebugEnabled())
			Log.debug("启动上下文事务....");
		synchronized (this) {
			Map<PlatformTransactionManager, TransactionStatus> txsMap = new LinkedHashMap<PlatformTransactionManager, TransactionStatus>();
			for (Map.Entry<String, DataSource> dsEns :  getRawDataSources().entrySet()) {
				if (Log.isDebugEnabled())
					Log.debug(String.format("启动数据源(%s)的数据库事务....", dsEns.getKey()));
				try {
					PlatformTransactionManager tx = new DataSourceTransactionManager(dsEns.getValue());
					TransactionStatus txStatus = tx.getTransaction(new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRES_NEW));
					txsMap.put(tx, txStatus);
					if (Log.isDebugEnabled())
						Log.debug(String.format("启动数据源(%s)的数据库事务成功。", dsEns.getKey()));
				} catch (Throwable e) {
					Log.error(String.format("启动数据源(%s)的数据库事务失败！", dsEns.getKey()));
					throw new IllegalStateException(e.getMessage(), e);
				}
			}
			transactionStatuses = new HashMap<PlatformTransactionManager, TransactionStatus>(txsMap);
		}
	}
	
	public void commitContextTransaction() {
		if (transactionStatuses == null) return;
		synchronized (this) {
			if (Log.isDebugEnabled())
				Log.debug("正在提交上下文事务....");
			try {
				for (Map.Entry<PlatformTransactionManager, TransactionStatus> txEns : transactionStatuses.entrySet()) {
						txEns.getKey().commit(txEns.getValue());
				}
			} catch (Throwable e) {
				Log.error(String.format("提交上下文的数据库事务失败！"));
				throw new IllegalStateException(e.getMessage(), e);
			}
		}
	}

	public void rollbackContextTransaction() {
		if (transactionStatuses == null) return;
		synchronized (this) {
			if (Log.isDebugEnabled())
				Log.debug("正在回滚上下文事务....");
			for (Map.Entry<PlatformTransactionManager, TransactionStatus> txEns : transactionStatuses.entrySet()) {
				if (!txEns.getValue().isCompleted()) {
					txEns.getKey().rollback(txEns.getValue());
				}
			}
		}
	}
	
	public DataSourceFactory getDatasourceFactory() {
		return dataSourceFactory;
	}
	
	public MessageProcessor getMessageProcessor() {
		return messageProcessor;
	}
	
	public TransformReactor getTransformReactor() {
		return transformReactor;
	}
	
}
